package com.xf.day07_graphx
import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.graphx.{GraphLoader, PartitionStrategy}
import org.apache.spark.rdd.RDD

object Triangle {
  def main(args: Array[String]): Unit = {

    Logger.getLogger("org").setLevel(Level.OFF)
    System.setProperty("spark.ui.showConsoleProgress", "false")
    //运行环境
    val sconf = new SparkConf().setMaster("local[4]").setAppName("MLlib")
    val sc = new SparkContext(sconf)
    sc.setLogLevel("error")

    //followers.txt 文件中====>  第一列为用户名，第二列为社区编号。
    val graph = GraphLoader
      .edgeListFile(sc, "file:///D:\\workRecord\\spark_project20250924\\data\\SimpleGraphX\\followers.txt", true)
      .partitionBy(PartitionStrategy.RandomVertexCut)

    //通过顶点进行社区发现
    // 对图进行三角形计数
    val triCounts = graph.triangleCount().vertices


    /**
    * 三角形计数的结果
    *    (4,0)
    *    (1,0)
    *    (6,1)
    *    (3,1)
    *    (7,1)
    *    (2,0)
    */
    println("=============================================>")
    triCounts.collect().foreach( println)

    //读取用户数据 ====> 第一列为 用户的编号    第二列为 昵称
    val users = sc.textFile("file:///D:\\workRecord\\spark_project20250924\\data\\SimpleGraphX\\users.txt")
      .map { line =>
      val fields = line.split(",")
      (fields(0).toLong, fields(1))
    }

    //对发现的结果与用户数据进行关联  ===>  ( 用户的编号,  ( 昵称,  次数 ) )
    val triCountByUsername :RDD[(String, Int)]= users.join(triCounts).map { case (id, (username, tc)) => (username, tc) }

    /**
     * 注意:
     *    (justinbieber,0)
     *    (matei_zaharia,1)
     *    (ladygaga,0)
     */
    println("=============================================>")
    triCountByUsername.take(3).foreach(println)

    println("=============================================>")

    //打印发现社区中包含的成员数  --->  三角形计数   对应的顶点个数
    triCountByUsername.map(x => (x._2, 1)).reduceByKey(_ + _).foreach {
      case (communication, count) => println("The Communication: " + communication + " have " + count + " members.")
    }

    //按照社区的顺序，打印详细信息  ( 三角形计数, 用户昵称 )
    println("All communication's members:   ===========================> ")
    println( triCountByUsername.map(x => (x._2, x._1)).sortByKey(true).collect().mkString("\n"))

  }
}